package com.xzx.coprocessor;

import lombok.extern.java.Log;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessor;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.coprocessor.RegionObserver;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.wal.WALEdit;

import java.io.IOException;
import java.util.List;
import java.util.Optional;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/8/10 14:56
 */
@Slf4j
public class CallLogCoprocessor implements RegionObserver, RegionCoprocessor {
    // 初始化
    @Override
    public void start(CoprocessorEnvironment env) throws IOException {
        RegionCoprocessor.super.start(env);
        log.info("====协处理器加载。每个region上面都会被执行一次，不是每个regionserver");
    }

    // 销毁
    @Override
    public void stop(CoprocessorEnvironment env) throws IOException {
        RegionCoprocessor.super.stop(env);
        log.info("====协处理器卸载");
    }
    /**
	 * 加入该方法，否则无法生效,xzx 重要的方法
	 */
	@Override
	public Optional<RegionObserver> getRegionObserver() {
		return Optional.of(this);
	}

    @Override
    public void postPut(ObserverContext<RegionCoprocessorEnvironment> c, Put put, WALEdit edit) throws IOException {
        RegionObserver.super.postPut(c, put, edit);
        log.info("=== postPut:{}", put.toJSON());
        try(Table table = c.getEnvironment().getConnection().getTable(TableName.valueOf("xzx:call_log"))){
            List<Cell> cells = put.get("d".getBytes(), "isCaller".getBytes());
            log.info("=== cell size:{}", cells.size());
            if (CollectionUtils.isEmpty(cells)) {
                return;
            }
            String isCaller = Bytes.toString(CellUtil.cloneValue(cells.get(0)));
            log.info("=====isCaller:{}", isCaller);
            if (!"1".equals(isCaller)) {
                return;
            }
            byte[] caller = CellUtil.cloneValue(put.get("d".getBytes(), "caller".getBytes()).get(0));
            byte[] callee = CellUtil.cloneValue(put.get("d".getBytes(), "callee".getBytes()).get(0));
            byte[] callTs = CellUtil.cloneValue(put.get("d".getBytes(), "callTs".getBytes()).get(0));
            byte[] duration = CellUtil.cloneValue(put.get("d".getBytes(), "duration".getBytes()).get(0));

            int regionNumber = (Bytes.toString(callee) + Bytes.toString(callTs).substring(0, 6)).hashCode() % 3;
            String newRowKey = regionNumber + "|" + Bytes.toString(callee) + "_" + Bytes.toString(callTs);
            Put newPut = new Put(newRowKey.getBytes());
            newPut.addColumn("d".getBytes(), "caller".getBytes(), callee);
            newPut.addColumn("d".getBytes(), "callee".getBytes(), caller);
            newPut.addColumn("d".getBytes(), "callTs".getBytes(), callTs);
            newPut.addColumn("d".getBytes(), "duration".getBytes(), duration);
            newPut.addColumn("d".getBytes(), "isCaller".getBytes(), "0".getBytes());
            table.put(newPut);
        } catch (Exception e) {
            log.error("call log coprocessor error", e);
        }
    }
}